package com.zhang.spark_2.com.zhang.core.req

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

import java.text.SimpleDateFormat
import java.util.Date

/**
 * @title: 页面平均访问时间
 * @author: zhang
 * @date: 2022/2/16 17:16 
 */
object Spark04_req {

  def main(args: Array[String]): Unit = {
    //TODO 获取环境
    val conf = new SparkConf().setMaster("local").setAppName("WordCount")
    val sc = new SparkContext(conf);
    val userVisit: RDD[String] = sc.textFile("data/user_visit_action.txt")

    val mapObjs: RDD[UserVisitAction] = userVisit.map(
      line => {
        val data: Array[String] = line.split("_")
        UserVisitAction(
          data(0),
          data(1).toLong,
          data(2),
          data(3).toLong,
          data(4),
          data(5),
          data(6).toLong,
          data(7).toLong,
          data(8),
          data(9),
          data(10),
          data(11),
          data(12).toLong
        )
      }
    )


    //todo 每个页面访问时间
    val zipDatas: RDD[(String, List[(Long, Long)])] = mapObjs
      .groupBy(_.session_id)
      .mapValues(
        iter => {
          val sortActions: List[UserVisitAction] = iter.toList.sortBy(_.action_time)
          val page_time: List[(Long, String)] = sortActions.map(data => (data.page_id, data.action_time))
          val tuples: List[((Long, String), (Long, String))] = page_time.zip(page_time.tail)
          val sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss")
          val pidTime: List[(Long, Long)] = tuples.map {
            case ((lastPid, lastTime), (nowPid, nowTime)) => {
              val lt: Date = sdf.parse(lastTime)
              val nt: Date = sdf.parse(nowTime)
              (lastPid, (nt.getTime - lt.getTime) / 1000)
            }
          }
          pidTime
        }
      )

    val pageRes: RDD[(Long, Long)] = zipDatas.map(_._2).flatMap(list => list)

    pageRes.cache()

    //每个页面访问总时长
    val pidSumTime: Map[Long, Long] = pageRes.reduceByKey(_ + _).collect().toMap


    //todo 计算每个页面访问的次数
    val pidCount: Map[Long, Int] = pageRes.map {
      case (pid, time) => {
        (pid, 1)
      }
    }.reduceByKey(_ + _)
      .collect()
      .toMap

    pidCount.map {
      case (pid, cnt) => {
        (pid, (pidSumTime.getOrElse(pid, 0L).toDouble / cnt))
      }
    }.foreach(println)


    sc.stop()
  }


  //用户访问动作表
  case class UserVisitAction(
                              date: String, //用户点击行为的日期
                              user_id: Long, //用户的ID
                              session_id: String, //Session的ID
                              page_id: Long, //某个页面的ID
                              action_time: String, //动作的时间点
                              search_keyword: String, //用户搜索的关键词
                              click_category_id: Long, //某一个商品品类的ID
                              click_product_id: Long, //某一个商品的ID
                              order_category_ids: String, //一次订单中所有品类的ID集合
                              order_product_ids: String, //一次订单中所有商品的ID集合
                              pay_category_ids: String, //一次支付中所有品类的ID集合
                              pay_product_ids: String, //一次支付中所有商品的ID集合
                              city_id: Long //城市 id
                            )

}
